package com.it.gmall.realtime.util;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.spark.Partition;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.apache.spark.streaming.kafka010.PerPartitionConfig;

import java.util.*;

/**
 * @author : code1997
 * @date :2021-01-2021/1/27 23:35
 */
public class KafkaUtil {

    private static Map<String, Object> kafkaParams = null;

    static {
        Properties properties = PropertiesUtil.load("kafka.properties");
        if (kafkaParams == null) {
            synchronized (KafkaUtil.class) {
                kafkaParams = new HashMap<>();
                kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.get("kafka.broker.list"));
                kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, properties.get("group.id"));
                kafkaParams.put("key.deserializer", properties.get("key.deserializer"));
                kafkaParams.put("value.deserializer", properties.get("value.deserializer"));
                kafkaParams.put("auto.offset.reset", properties.get("auto.offset.reset"));
                kafkaParams.put("enable.auto.commit", properties.get("enable.auto.commit"));
            }
        }
    }

    public static JavaInputDStream<ConsumerRecord<String, String>> getKafKaStream(String topic, JavaStreamingContext jsc) {
        return KafkaUtils.createDirectStream(
                jsc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(Collections.singletonList(topic), kafkaParams));
    }

    public static JavaInputDStream<ConsumerRecord<String, String>> getKafKaStream(String topic, JavaStreamingContext jsc, String groupId) {
        kafkaParams.put("ConsumerConfig.GROUP_ID_CONFIG", groupId);
        return KafkaUtils.createDirectStream(
                jsc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(Collections.singletonList(topic), kafkaParams));
    }

    public static JavaInputDStream<ConsumerRecord<String, String>> getKafKaStream(String topic, JavaStreamingContext jsc, String groupId, Map<TopicPartition, Long> offset) {
        kafkaParams.put("ConsumerConfig.GROUP_ID_CONFIG", groupId);
        return KafkaUtils.createDirectStream(
                jsc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(Collections.singletonList(topic), kafkaParams, offset));
    }


}
